package bsp;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hama.bsp.BSP;
import org.apache.hama.bsp.BSPPeer;
import org.apache.hama.bsp.sync.SyncException;

import util.LongPair;
import util.functors.OpScan;

public abstract class BspHelper<IK, IV, OK, OV, MS extends Writable> extends
		BSP<IK, IV, OK, OV, MS> {

	protected ArrayList<CMapper> mappers;
	protected CReducer<?, ?, ?, ?, ?> reducer;

	public BspHelper() {
		mappers = new ArrayList<CMapper>();
		reducer = null;
	}

	public void addMapper(CMapper<IK, IV, OK, OV, MS> mapper) {
		// System.out.println("*****set a map*****");
		if (mappers == null)
			mappers = new ArrayList<CMapper>();
		mappers.add(mapper);
	}

	public abstract void setAllCMappers();

	public void addReducer(CReducer<IK, IV, OK, OV, MS> reducer) {

		this.reducer = reducer;
	}

	@Override
	public void setup(BSPPeer<IK, IV, OK, OV, MS> peer) throws IOException,
			SyncException, InterruptedException {
		setAllCMappers();
	}

	@Override
	public void bsp(BSPPeer<IK, IV, OK, OV, MS> peer) throws IOException,
			SyncException, InterruptedException {
		for (CMapper<IK, IV, OK, OV, MS> mapper : mappers) {
			// System.out.println("*****map*****");
			mapper.map(peer);
			mapper.sendAllMesgs(peer);
			// System.out.println("*****send*****");
			peer.reopenInput();
			peer.sync();
		}

	}

}
